package com.tensai.study.新概念.响应式编程.reactor.flux;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @author Tensai
 */
public class _Test {
    @Test
    public void compositeOperate() throws InterruptedException {
        Set<Integer> set = new HashSet<>();
        set.add(1);
        set.add(2);
        set.add(3);
        Flux.fromIterable(set)
                .delayElements(Duration.ofMillis(1))
                .doOnNext(this::doSomeThing) //the callback to call on {@link Subscriber#onNext}
                .publishOn(Schedulers.elastic())
                .map(d -> d * 2)
                .take(3)
                .doOnCancel(() -> System.out.println("操作被取消了"))
                .doOnComplete(() -> System.out.println("正常结束了"))
                .onErrorResume(e -> e instanceof RuntimeException, e -> {
                    System.out.println("出现了 RuntimeException");
                    return Flux.error(e);
                })//某种异常时才处理
                .doOnError(e -> System.out.println("出现了 Exception"))
                .doAfterTerminate(() -> System.out.println("不管正常异常，反正是结束了"))
                .onErrorReturn(444)
                .doFinally(signalType -> System.out.println("finally--->" + signalType))
                .subscribe(System.out::println);

        TimeUnit.SECONDS.sleep(5);
    }


    public Integer doSomeThing(Integer integer) {
        //自行调节看效果
        if (1 != 1) {
            throw new RuntimeException("a RuntimeException error");
        } else if (2 != 2) {
            throw new NullPointerException("a NullPointerException error");
        }
        return integer + 1;
    }


}
